package online.heycm.platform.rocketmq.publisher;

import lombok.extern.slf4j.Slf4j;
import online.heycm.platform.common.entity.exception.Assert;
import online.heycm.platform.rocketmq.common.entity.Constant;
import online.heycm.platform.rocketmq.common.entity.Event;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;

import java.util.function.Function;

/**
 * @author heycm
 * @since 2023/12/2 20:40
 */
@Slf4j
public class EventPublisherImpl implements EventPublisher {

    private final RocketMQTemplate rocketMQTemplate;

    public EventPublisherImpl(RocketMQTemplate rocketMQTemplate, TransactionListener transactionListener) {
        this.rocketMQTemplate = rocketMQTemplate;
        TransactionMQProducer producer = (TransactionMQProducer) rocketMQTemplate.getProducer();
        producer.setTransactionListener(transactionListener);
    }

    private boolean success(SendResult result) {
        Assert.notNull(result, "SendResult is not allowed null");
        if (result instanceof TransactionSendResult) {
            TransactionSendResult ts = (TransactionSendResult) result;
            return SendStatus.SEND_OK == ts.getSendStatus() && LocalTransactionState.COMMIT_MESSAGE == ts.getLocalTransactionState();
        }
        return SendStatus.SEND_OK == result.getSendStatus();
    }

    private boolean sendOperator(SendOperator operator) {
        try {
            SendResult result = operator.apply();
            boolean success = success(result);
            if (success) {
                log.info("{}", result);
            } else {
                log.error("{}", result);
            }
            return success;
        } catch (Exception e) {
            log.error("消息推送异常: {}", e.getMessage(), e);
        }
        return false;
    }

    @Override
    public boolean pushTransaction(Event event) {
        LocalTransactionHandlerRegistry.register(event);
        return sendOperator(() -> rocketMQTemplate.sendMessageInTransaction(event.getEventTopic(),
                MessageBuilder.withPayload(event).build(), "")
        );
    }

    @Override
    public boolean pushBroadcast(Event event) {
        LocalTransactionHandlerRegistry.register(event);
        return sendOperator(() ->
                rocketMQTemplate.sendMessageInTransaction(event.getEventTopic() + Constant.BROADCAST_SUFFIX,
                        MessageBuilder.withPayload(event).build(), "")
        );
    }

    @Override
    public boolean pushDelay(Event event) {
        return sendOperator(() -> rocketMQTemplate.syncSend(event.getEventTopic(),
                MessageBuilder.withPayload(event).build(), event.getTimeout(), event.getDelay())
        );
    }

    @Override
    public boolean pushSync(Event event) {
        return sendOperator(() -> rocketMQTemplate.syncSend(event.getEventTopic(),
                MessageBuilder.withPayload(event).build(), event.getTimeout())
        );
    }

    @Override
    public void pushAsync(Event event, SendCallback sendCallback) {
        rocketMQTemplate.asyncSend(event.getEventTopic(), MessageBuilder.withPayload(event).build(), sendCallback,
                event.getTimeout());
    }

    @Override
    public boolean pushOrderly(Event event) {
        return sendOperator(() -> rocketMQTemplate.syncSendOrderly(event.getEventTopic() + Constant.ORDERLY_SUFFIX,
                MessageBuilder.withPayload(event).build(), event.getHash(), event.getTimeout())
        );
    }
}
